package com.ndp.fb.mq.producer.helper;

import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.selector.SelectMessageQueueByDataCenter;
import com.alibaba.rocketmq.common.message.Message;
import com.ndp.fb.config.ConfigCentre;
import com.ndp.fb.constant.ConfigConst;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;

/**
 * Created by infi on 2015/1/27.
 */
@Component
public class RocketMqProducerHelper implements InitializingBean, DisposableBean {
    private static Logger logger = LoggerFactory.getLogger(RocketMqProducerHelper.class);

    private String PRODUCER_GROUP = ConfigConst.MQ_SERVICES_PRODUCER_GROUP;
    private String ROCKETMQ_NAMESRV_DOMAIN = ConfigCentre.getString(ConfigConst.MQ_SERVICES_ROCKETMQ_NAMESRV_DOMAIN);
    private String MQ_SERVICES_ROCKET_MQCLIENT_PASSWORD = ConfigCentre.getString(ConfigConst.MQ_SERVICES_ROCKET_MQCLIENT_PASSWORD);
    private String MQ_SERVICES_LOG_HOME = ConfigCentre.getString(ConfigConst.MQ_SERVICES_LOG_HOME);
    private String ROCKET_MQ_DATACENTER_INDEX=ConfigCentre.getString(ConfigConst.MQ_SERVICES_DATACENTER_INDEX);

    private SelectMessageQueueByDataCenter selectMessageQueueByDataCenter;
    private DefaultMQProducer producer;

    @Override
    public void destroy() throws Exception {
        producer.shutdown();

    }

    @Override
    public void afterPropertiesSet() throws Exception {
        try {
            System.setProperty(ConfigConst.MQ_SERVICES_ENABLE_SSL, ConfigCentre.getString(ConfigConst.MQ_SERVICES_ENABLE_SSL));
            if (!StringUtils.isBlank(ROCKETMQ_NAMESRV_DOMAIN)) {
                System.setProperty(ConfigConst.MQ_SERVICES_ROCKETMQ_NAMESRV_DOMAIN, ROCKETMQ_NAMESRV_DOMAIN);
            }
            if (!StringUtils.isBlank(MQ_SERVICES_ROCKET_MQCLIENT_PASSWORD)) {
                System.setProperty(ConfigConst.MQ_SERVICES_ROCKET_MQCLIENT_PASSWORD, MQ_SERVICES_ROCKET_MQCLIENT_PASSWORD);
            }
            if (!StringUtils.isBlank(MQ_SERVICES_LOG_HOME)) {
                System.setProperty(ConfigConst.MQ_SERVICES_LOG_HOME, MQ_SERVICES_LOG_HOME);
            }
            if (!StringUtils.isBlank(ROCKET_MQ_DATACENTER_INDEX)) {
                System.setProperty(ConfigConst.MQ_SERVICES_DATACENTER_INDEX, ROCKET_MQ_DATACENTER_INDEX);
            }

            logger.info("准备构造消生产者 className >>>> " + this.getClass().getName());
            logger.info("producerGroup >>>>> " + PRODUCER_GROUP);
            logger.info("rocketmq.namesrv.domain >>> " + System.getProperty(ConfigConst.MQ_SERVICES_ROCKETMQ_NAMESRV_DOMAIN));
            logger.info("enable_ssl >>> " + System.getProperty(ConfigConst.MQ_SERVICES_ENABLE_SSL));
            logger.info("log.home >>> " + System.getProperty(ConfigConst.MQ_SERVICES_LOG_HOME));
            producer = new DefaultMQProducer(PRODUCER_GROUP);
            producer.start();
            selectMessageQueueByDataCenter = new SelectMessageQueueByDataCenter(producer);
        } catch (MQClientException e) {
            logger.error("初始化生产者失败", e);
        }
    }

    public void send(Message msg) throws Exception {
        if (producer == null) {
            logger.error("消息生成者未创建");
        } else {
            producer.send(msg, selectMessageQueueByDataCenter, null);
        }
    }
}
